package cn.zzdt4j.config.springboot.starter.refresher;

import cn.hutool.core.util.StrUtil;
import cn.zzdt4j.threadpool.dynamic.model.config.properties.BootstrapConfigProperties;
import cn.zzdt4j.threadpool.message.core.service.GlobalNotifyAlarmManage;
import cn.zzdt4j.threadpool.message.core.service.ThreadPoolNotifyAlarm;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.CuratorEvent;
import org.apache.curator.framework.api.CuratorListener;
import org.apache.curator.framework.api.GetChildrenBuilder;
import org.apache.curator.framework.api.GetDataBuilder;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.utils.ZKPaths;
import org.apache.zookeeper.WatchedEvent;

import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Map;
import java.util.Objects;

/**
 * Zookeeper refresh handler
 *
 * @author by <a href="mailto:ligang941012@gmail.com">gang.Li</a>
 * @since 2024/5/13 15:46
 */
@Slf4j
public class ZookeeperRefresherHandler extends AbstractConfigThreadPoolDynamicRefresh {

    private CuratorFramework curatorFramework;

    private static final String CONFIG_VERSION = "config_version";

    private static final String ROOT_NODE = "root_node";

    private static final String NODE = "node";

    private static final String ZK_CONNECT_STR = "zk_connect_str";

    private static final int BASE_SLEEP_TIME_MS = 1000;

    private static final int MAX_RETRIES = 3;

    @Override
    public void registerListener() {
        BootstrapConfigProperties actualBootstrapConfigProperties = (BootstrapConfigProperties) bootstrapConfigProperties;
        final Map<String, String> zkConfigs = actualBootstrapConfigProperties.getZookeeper();
        curatorFramework = CuratorFrameworkFactory.newClient(zkConfigs.get(ZK_CONNECT_STR),
                new ExponentialBackoffRetry(BASE_SLEEP_TIME_MS, MAX_RETRIES));
        final String nodePath = ZKPaths.makePath(ZKPaths.makePath(zkConfigs.get(ROOT_NODE),
                zkConfigs.get(CONFIG_VERSION)), zkConfigs.get(NODE));
        final ConnectionStateListener connectionStateListener = (client, newState) -> {
            if (newState == ConnectionState.CONNECTED) {
                loadNode(nodePath);
            } else if (newState == ConnectionState.RECONNECTED) {
                loadNode(nodePath);
            }
        };

        final CuratorListener curatorListener = (client, curatorEvent) -> {
            final WatchedEvent watchedEvent = curatorEvent.getWatchedEvent();
            if (!Objects.isNull(watchedEvent)) {
                switch (watchedEvent.getType()) {
                    case NodeChildrenChanged:
                    case NodeDataChanged:
                        loadNode(nodePath);
                        break;
                    default:
                        break;
                }
            }
        };
        curatorFramework.getConnectionStateListenable().addListener(connectionStateListener);
        curatorFramework.getCuratorListenable().addListener(curatorListener);
        curatorFramework.start();
    }

    private void loadNode(String nodePath) {
        final String content = nodePathResolver(nodePath);
        if (StrUtil.isNotBlank(content)) {
            dynamicRefresh(content);
            registerNotifyAlarmManager();
        }
    }

    private String nodePathResolver(String nodePath) {
        try {
            final GetChildrenBuilder childrenBuilder = curatorFramework.getChildren();
            final List<String> children = childrenBuilder.watched().forPath(nodePath);
            final StringBuilder content = new StringBuilder();
            children.forEach(each -> {
                final String nodeName = ZKPaths.getNodeFromPath(each);
                final GetDataBuilder data = curatorFramework.getData();
                String value = "";
                try {
                    value = new String(data.watched().forPath(each), StandardCharsets.UTF_8);
                } catch (Exception ex) {
                    log.error("Load zookeeper node error", ex);
                }
                content.append(nodeName).append("=").append(value).append("\n");
            });
            return content.toString();
        } catch (Exception e) {
            log.error("Load zookeeper node error, nodePath is: {}", nodePath, e);
            return null;
        }
    }

    /**
     * Registers notification alarm managers for each configured executor.
     * Iterates through executor properties, initializing and configuring a ThreadPoolNotifyAlarm
     * for each, then registers these alarms with the global notification alarm manager.
     * This process includes parsing bootstrap configuration properties and setting up notification
     * configurations for individual executors.
     */
    public void registerNotifyAlarmManager() {
        // Obtain the actual bootstrap configuration properties via type casting.
        BootstrapConfigProperties actualBootstrapConfigProperties = (BootstrapConfigProperties) bootstrapConfigProperties;
        // Iterate over all executor configurations to setup notification alarms accordingly.
        actualBootstrapConfigProperties.getExecutors().forEach(executorProperties -> {
            // Instantiate a ThreadPoolNotifyAlarm with specified alarm triggers.
            final ThreadPoolNotifyAlarm threadPoolNotifyAlarm = new ThreadPoolNotifyAlarm(
                    executorProperties.getAlarm(),
                    executorProperties.getCapacityAlarm(),
                    executorProperties.getActiveAlarm()
            );
            // Configure the check interval and recipients list for the alarm.
            threadPoolNotifyAlarm.setInterval(executorProperties.getNotify().getInterval());
            threadPoolNotifyAlarm.setReceives(executorProperties.getNotify().getReceives());
            // Register the created ThreadPoolNotifyAlarm with the global alarm manager.
            GlobalNotifyAlarmManage.put(executorProperties.getThreadPoolId(), threadPoolNotifyAlarm);
        });
    }

}
